package com.huatai.datacenter.job;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.toolkit.SqlRunner;
import com.huatai.datacenter.entity.DataServicesEntity;
import com.huatai.datacenter.entity.datamodel.DataModelFieldEntity;
import com.huatai.datacenter.service.ClusterService;
import com.huatai.datacenter.service.DataServices;
import com.huatai.datacenter.service.KafkaConsumerService;
import com.huatai.datacenter.service.RequestService;
import com.huatai.datacenter.service.impl.DataModelFieldServiceImpl;
import com.huatai.datacenter.service.impl.QualityRuleServiceImpl;
import com.huatai.datacenter.websocket.WebSocketServer;
import com.huatai.dataservice.dubbo.DataDisposeDubboAPI;
import com.huatai.dataservice.dubbo.DataDisposeDubboAPIMongoDB;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.dubbo.config.annotation.DubboReference;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.sql.Statement;
import java.time.LocalDateTime;
import java.util.Iterator;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.stream.Collectors;

/**
 * @author Lion
 * @date 2023/6/13  9:22
 */
@Service
@Slf4j
public class KafkaConsumerListener {

	@Autowired
	private DataModelFieldServiceImpl dataModelFieldService;

	@Autowired
	private QualityRuleServiceImpl qualityRuleService;

	@Autowired
	private KafkaConsumerService kafkaConsumerService;

	@Autowired
	private ClusterService clusterService;

	@Autowired
	private DataServices dataServices;

	@Autowired
	private RequestService requestService;

	@DubboReference
	private DataDisposeDubboAPI dataDisposeDubbo;

	// 存到MongoDB的
	@DubboReference
	private DataDisposeDubboAPIMongoDB dataDisposeDubboAPIMongoDB;


	/**
	 * kafka消息处理类
	 *
	 * @param consumerRecord
	 */
	public void listen(ConsumerRecord<String, String> consumerRecord, long modelId, String modelName, String tableName, String structure, String qualityRule) throws Exception {

		QueryWrapper<DataModelFieldEntity> dataModelWrapper = new QueryWrapper<>();
		dataModelWrapper.eq("data_model_id", modelId);
		List<String> list = dataModelFieldService.list(dataModelWrapper).stream().map(DataModelFieldEntity::getFieldName).collect(Collectors.toList());
		String sql = "";
		try {
			String value = (String) consumerRecord.value();

			// 根据模型id将数据获取保存到表
			JSONObject jsonObject = JSONObject.parseObject(value);
			if (StringUtils.isBlank(structure)){
				//质量规则校验
				qualityRuleData(modelId,modelName,qualityRule,jsonObject);
				//保存数据到数据库中
				sql = dataDisposeDubbo.generateInsertStatement(tableName,jsonObject,list);
			}else {
				//质量规则校验
				qualityRuleData(modelId,modelName,qualityRule,jsonObject.getJSONObject(structure));
				//保存数据到数据库中
				sql = dataDisposeDubbo.generateInsertStatement(tableName, jsonObject.getJSONObject(structure), list);
			}

			//如果在实时数据服务中有该数据主题，则发送数据
			String topicName = consumerRecord.topic();
			QueryWrapper<DataServicesEntity> queryWrapper = new QueryWrapper<>();
			queryWrapper.eq("targetTopic",topicName);
			List<DataServicesEntity>  data  = dataServices.list(queryWrapper);
			//判断主题是否在服务中,如果服务中有，则发送该数据  判断是通过kafka发送还是websocket发送
			if (data != null){
				Iterator<DataServicesEntity> iterator = data.iterator();
				if (iterator.hasNext()){
					DataServicesEntity dataServices = iterator.next();
					if ("kafka".equals(dataServices.getServiceType())){
						sendData(modelId,jsonObject,Long.valueOf(dataServices.getClusterId()),dataServices.getTopicName());
					}
					if ("websocket".equals(dataServices.getServiceType())){
						sendDataToWebSocket(jsonObject,dataServices.getTopicName());
					}
				}
			}
			//执行SQL，保存数据 不创建连接
			dataDisposeDubbo.insertDataToTable(sql);

		} catch (Exception e) {
			log.info("错误信息为："+e.getMessage());
		}
	}




	/**
	 * kafka消息处理类 存入MongoDB的
	 *
	 * @param consumerRecord 消费者记录
	 * @param modelId
	 * @param modelName
	 * @param tableName 表名
	 * @param structure key
	 * @param qualityRule 质量规则
	 * @throws Exception 异常
	 * @author lizhong
	 * @date 2023年8月1日
	 */
	public void listenMongoDB(ConsumerRecord<String, String> consumerRecord, long modelId, String modelName, String tableName, String structure, String qualityRule) throws Exception {

		QueryWrapper<DataModelFieldEntity> dataModelWrapper = new QueryWrapper<>();
		dataModelWrapper.eq("data_model_id", modelId);

		// 模型字段
		List<String> list = dataModelFieldService.list(dataModelWrapper).stream().map(DataModelFieldEntity::getFieldName).collect(Collectors.toList());
		try {
			String value = consumerRecord.value();

			// 根据模型id将数据获取保存到表，要保存到MongoDB的数据
			JSONObject jsonObject = JSONObject.parseObject(value);

			if (StringUtils.isBlank(structure)) {
				//质量规则校验
				qualityRuleData(modelId, modelName, qualityRule, jsonObject);
				//保存数据到数据库MongoDB中
				dataDisposeDubboAPIMongoDB.executeInsert(tableName, jsonObject);
			} else {
				//质量规则校验
				qualityRuleData(modelId, modelName, qualityRule,jsonObject.getJSONObject(structure));
				//保存数据到数据库MongoDB中
				dataDisposeDubboAPIMongoDB.executeInsert(tableName, jsonObject.getJSONObject(structure));
			}

			//如果在实时数据服务中有该数据主题，则发送数据
			String topicName = consumerRecord.topic();
			QueryWrapper<DataServicesEntity> queryWrapper = new QueryWrapper<>();
			queryWrapper.eq("targetTopic", topicName);
			List<DataServicesEntity> data = dataServices.list(queryWrapper);
			//判断主题是否在服务中,如果服务中有，则发送该数据  判断是通过kafka发送还是websocket发送
			if (data != null) {
				Iterator<DataServicesEntity> iterator = data.iterator();
				if (iterator.hasNext()) {
					DataServicesEntity dataServices = iterator.next();
					if ("kafka".equals(dataServices.getServiceType())) {
						sendData(modelId, jsonObject, Long.valueOf(dataServices.getClusterId()), dataServices.getTopicName());
					}
					if ("websocket".equals(dataServices.getServiceType())) {
						sendDataToWebSocket(jsonObject, dataServices.getTopicName());
					}
				}
			}

		} catch (Exception e) {
			log.info("错误信息为：" + e.getMessage());
		}
	}

	/**
	 * @author Lion
	 * @description 质量规则判断
	 * @param modelId
	 * @param modelName
	 * @param qualityRule
	 * @param jsonData
	 * @return void
	 */
	public void qualityRuleData(long modelId,String modelName,String qualityRule,JSONObject jsonData){

		if (StringUtils.isNoneBlank(qualityRule)){
			JSONArray jsonArray = JSONObject.parseArray(qualityRule);
			for (int i=0;i<jsonArray.size();i++){

				String filedName = jsonArray.getJSONObject(i).getString("ruleField");
				long ruleId = jsonArray.getJSONObject(i).getLong("quality");

				String dataSource = jsonData.get(filedName).toString();

				qualityRuleService.checkQualityRuleByFieldName(dataSource,ruleId,modelId,modelName,filedName,LocalDateTime.now(),null);

			}
		}
	}

	/**
	 * @author Lion
	 * @description 数据发送
	 * @param modelId
	 * @param data
	 * @param clusterId
	 * @param topicName
	 * @return void
	 */
	public void sendData(long modelId,JSONObject data,long clusterId,String topicName){

		String broker = clusterService.getBrokers(String.valueOf(clusterId));
		Properties conf = kafkaConsumerService.generatorProducerProps(broker);
		KafkaProducer<String, String> producer = new KafkaProducer<>(conf);

		//获取数据模型
		List<DataModelFieldEntity> list = dataModelFieldService.getDataModelFieldList(modelId);
		Iterator<DataModelFieldEntity> iterable = list.iterator();
		JSONObject sendData = new JSONObject();
		while (iterable.hasNext()){
			DataModelFieldEntity dataModelField = iterable.next();
			sendData.put(dataModelField.getFieldName(),data.get(dataModelField.getFieldName()));
		}
		try {
			//发送数据
			ProducerRecord<String, String> record = new ProducerRecord<>(topicName, sendData.toString());
			producer.send(record);
			//记录发送成功日志同时同步到redis中
			//requestService.saveLog(topicName,"success","实时数据服务",sendData.toString(),"");

		}catch (Exception e){
			//记录发送失败日志同时同步到redis中
			//requestService.saveLog(topicName,"error","实时数据服务",sendData.toString(),"");
		}
	}

	/**
	 * @author Lion
	 * @description 将实时数据写入到websocket 中
	 * @param data
	 * @param clientId
	 * @return void
	 */
	public void sendDataToWebSocket(JSONObject data,String clientId){

		CopyOnWriteArraySet<WebSocketServer> webSocketSet = WebSocketServer.webSocketSet;
		String path = "";

		Iterator<WebSocketServer> webSocketServerIterator =  webSocketSet.iterator();
		while (webSocketServerIterator.hasNext()){

			WebSocketServer webSocketServer = webSocketServerIterator.next();
			path = webSocketServer.getClientId();

			if (clientId.equals(path)){
				webSocketServer.sendMessage(webSocketServer.getSession(),data.toString());
			}
		}
	}
}
